package com.akka.demo.actor;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Cancellable;
import akka.actor.Terminated;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.akka.demo.model.ReadTemperature;
import com.akka.demo.model.RespondTemperature;
import com.akka.demo.model.TemperatureReading;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import scala.concurrent.duration.FiniteDuration;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * Created by Administrator on 2018/2/27.
 */
@Component
@Scope("prototype")
public class DeviceGroupQuery extends AbstractActor {

    static public class CollectionTimeout {
    }

    private LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);

    private Map<ActorRef, String> actorToDeviceId = new HashMap<>();

    private long requestId;

    private ActorRef requester;

    private FiniteDuration timeout;

    public DeviceGroupQuery(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) {
        this.actorToDeviceId = actorToDeviceId;
        this.requestId = requestId;
        this.requester = requester;
        this.timeout = timeout;
    }

    private Cancellable queryTimeoutTimer;

    private Map<String, TemperatureReading> repliesSoFar = new HashMap<>();

    private Set<ActorRef> stillWaiting;

    @Override
    public void preStart() {
        queryTimeoutTimer = context().system().scheduler().scheduleOnce(timeout, self(), new CollectionTimeout(), context().system().dispatcher(), self());
        log.info("DeviceGroupQuery actor started");
        if (actorToDeviceId.keySet().size() > 0) {
            for (Iterator it = actorToDeviceId.keySet().iterator(); it.hasNext(); ) {
                ActorRef deviceActor = (ActorRef) it.next();
                context().watch(deviceActor);
                deviceActor.tell(new ReadTemperature(0), self());
            }
        }
    }

    @Override
    public void postStop() {
        log.info("DeviceGroupQuery actor stopped");
        queryTimeoutTimer.cancel();
    }

    @Override
    public Receive createReceive() {
        return waitingForReplies(repliesSoFar, actorToDeviceId.keySet());
    }

    private Receive waitingForReplies(Map<String, TemperatureReading> repliesSoFar, Set<ActorRef> stillWaiting) {
        return receiveBuilder()
                .match(RespondTemperature.class, respondTemperature -> {
                    log.info("Temperature is {},requestId is {}", respondTemperature.getValue(), respondTemperature.getRequestId());
                    TemperatureReading tr = 0 != respondTemperature.getValue() ? new DeviceGroup.Temperature(respondTemperature.getValue()) : new DeviceGroup.TemperatureNotAvailable();
                    receivedResponse(sender(), tr, repliesSoFar, stillWaiting);
                })
                .match(Terminated.class, terminated -> {
                    receivedResponse(sender(), new DeviceGroup.DeviceNotAvailable(), repliesSoFar, stillWaiting);
                })
                .match(CollectionTimeout.class, collectionTimeout -> {
                    if (stillWaiting.size() > 0) {
                        Map<String, TemperatureReading> timedOutReplies = new HashMap<>();
                        for (Iterator it = stillWaiting.iterator(); it.hasNext(); ) {
                            String deviceId = actorToDeviceId.get(it.next());
                            timedOutReplies.put(deviceId, new DeviceGroup.DeviceTimedOut());
                        }
                        repliesSoFar.putAll(timedOutReplies);
                        requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, repliesSoFar), self());
                    }
                    context().stop(self()); //超时自我关闭
                })
                .build();
    }

    private void receivedResponse(ActorRef deviceActor, TemperatureReading reading, Map<String, TemperatureReading> repliesSoFar, Set<ActorRef> stillWaiting) {
        context().unwatch(deviceActor); //不需要在收到温度消息后再收到Terminated事件，解除观察
        String deviceId = actorToDeviceId.get(deviceActor);
        stillWaiting.remove(deviceActor);
        repliesSoFar.put(deviceId, reading);
        if (stillWaiting.isEmpty()) {
            requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, repliesSoFar), self());
            context().stop(self());
        } else {
            context().become(waitingForReplies(repliesSoFar, stillWaiting).onMessage());
        }
    }
}
